/*
* This file is a part of the open-eBackup project.
* This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
* If a copy of the MPL was not distributed with this file, You can obtain one at
* http://mozilla.org/MPL/2.0/.
*
* Copyright (c) [2024] Huawei Technologies Co.,Ltd.
*
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
*/
#include <yaml-cpp/yaml.h>
#include <set>
#include <log/Log.h>
#include <unordered_map>
#include <boost/algorithm/string/trim.hpp>
#include <boost/algorithm/string/replace.hpp>
#include <boost/beast.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/ssl.hpp>
#include <boost/beast/websocket.hpp>
#include <boost/beast/websocket/ssl.hpp>
#include <boost/asio/connect.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/ssl/stream.hpp>
#include "client/KubeClient.h"
#include "curl_http/HttpStatus.h"
#include "protect_engines/kubernetes/common/KubeCommonInfo.h"
#include "protect_engines/kubernetes/common/KubeErrorCodes.h"
#include "KubernetesApi.h"

using Module::HttpRequest;

namespace KubernetesPlugin {
namespace {
const std::string MODULE_NAME = "KubernetesApi";

std::vector<std::string> LoadVolumesNames(const YAML::Node &volumeItems)
{
    std::vector<std::string> volumeNames = {};
    for (YAML::const_iterator vIt = volumeItems.begin(); vIt != volumeItems.end(); vIt++) {
        const YAML::Node &volumeClaim = *vIt;
        auto storageClassName = volumeClaim["spec"]["storageClassName"].as<std::string>();
        if (storageClassName.rfind("storage", 0) == 0) {
            auto volumeName = volumeClaim["metadata"]["name"].as<std::string>();
            volumeNames.push_back(volumeName);
        }
    }
    return volumeNames;
}
}

namespace Beast = boost::beast;         // from <boost/beast.hpp>
namespace Http = Beast::http;           // from <boost/beast/http.hpp>
namespace Websocket = Beast::websocket; // from <boost/beast/websocket.hpp>
namespace Net = boost::asio;            // from <boost/asio.hpp>
namespace Ssl = boost::asio::ssl;       // from <boost/asio/ssl.hpp>
using Tcp = boost::asio::ip::tcp;       // from <boost/asio/ip/tcp.hpp>
using Defer = std::shared_ptr<void>;

std::pair<int, std::vector<ApplicationResource>> KubernetesPlugin::KubernetesApi::ListNameSpaces()
{
    std::vector<ApplicationResource> namespaceVector = {};
    std::string url = config.GetClusterServer() + "/apis/apps/v1/statefulsets";

    auto [ret, statefulSetBody] = GetHttpBodyFromFlexVolume(url);
    if (ret != SUCCESS) {
        return std::make_pair(ret, namespaceVector);
    }

    url = config.GetClusterServer() + "/api/v1/namespaces";
    auto [nRet, nameSpaceBody] = GetHttpBodyFromFlexVolume(url);
    if (nRet != SUCCESS) {
        return std::make_pair(nRet, namespaceVector);
    }

    try {
        namespaceVector = GetNameSpacesFromHttpBody(statefulSetBody, nameSpaceBody);
    }
    catch (YAML::InvalidNode &e) {
        ERRLOG("InvalidNode msg: %s", e.what());
        return std::make_pair(CONNECT_FAILED, namespaceVector);
    }
    catch (YAML::ParserException &e) {
        ERRLOG("Wrong body message: %s", nameSpaceBody.c_str());
        return std::make_pair(CONNECT_FAILED, namespaceVector);
    }
    return std::make_pair(ret, namespaceVector);
}

std::pair<int, std::vector<ApplicationResource>> KubernetesPlugin::KubernetesApi::ListPods(const Application &sft)
{
    int ret = SUCCESS;
    std::string podsHttpBody;
    std::string url = config.GetClusterServer() + "/api/v1/namespaces/" + sft.parentName + "/pods";
    std::tie(ret, podsHttpBody) = GetHttpBodyFromFlexVolume(url);
    std::vector<ApplicationResource> podVector = {};
    if (ret != SUCCESS) {
        return std::make_pair(ret, podVector);
    }
    YAML::Node nodeItems = YAML::Load(podsHttpBody)["items"];
    for (YAML::const_iterator nIt = nodeItems.begin(); nIt != nodeItems.end(); nIt++) {
        const YAML::Node &nodeItem = *nIt;
        auto podName = nodeItem["metadata"]["name"].as<std::string>();
        if (std::string::npos == podName.rfind(sft.name, 0)) {
            continue;
        }
        YAML::Node containers = nodeItem["spec"]["containers"];
        if (!containers) {
            ERRLOG("There is no any container in the POD, name: %s", podName.c_str());
            continue;
        }

        if (containers.size() == 0) {
            ERRLOG("Containers'size is 0 in the POD, name: %s", podName.c_str());
            continue;
        }

        PodExtendInfo podExtendInfo;
        for (const auto &container: containers) {
            auto containerName = container["name"].as<std::string>();
            podExtendInfo.m_containerNames.push_back(containerName);
        }

        ApplicationResource applicationResource;
        applicationResource.name = podName;
        applicationResource.parentName = sft.parentName;

        std::string podExtendInfoStr;
        if (!Module::JsonHelper::StructToJsonString(podExtendInfo, podExtendInfoStr)) {
            ERRLOG("Struct PodExtendInfo to std::string fail.");
            continue;
        }
        applicationResource.extendInfo = podExtendInfoStr;
        podVector.push_back(applicationResource);
    }
    return std::make_pair(ret, podVector);
}

std::pair<int, std::vector<Pv>> KubernetesPlugin::KubernetesApi::ListPvs(const ApplicationResource &pod,
                                                                         const std::vector<std::string> &volumeNames)
{
    int ret = SUCCESS;
    std::vector<Pv> pvList = {};
    try {
        std::tie(ret, pvList) = GetPvcList(pod, volumeNames);
        if (ret != SUCCESS) {
            ERRLOG("Fail to parse pvc name list.");
            return std::make_pair(ret, pvList);
        }

        std::tie(ret, pvList) = GetPvsByPvcNames(pvList);
        if (ret != SUCCESS) {
            ERRLOG("Fail to parse pv list.");
            return std::make_pair(ret, pvList);
        }
    } catch (std::exception &e) {
        ERRLOG("Fail to parse pv list, message: %s", e.what());
        ret = FAILED;
    }
    for (const auto &item: pvList) {
        INFOLOG("There is a pv be found, pvc name: %s, lun name: %s in storage: %s",
            item.name.c_str(), item.lunName.c_str(), item.storageUrl.c_str());
    }
    return std::make_pair(ret, pvList);
}

std::pair<int32_t, std::vector<Pv>> KubernetesPlugin::KubernetesApi::GetPvcList(
    const ApplicationResource &pod, const std::vector<std::string> &volumeNames)
{
    int ret = SUCCESS;
    std::vector<Pv> pvcVector = {};
    std::string podHttpBody;
    std::string url = config.GetClusterServer() + "/api/v1/namespaces/" + pod.parentName + "/pods/" + pod.name;
    std::tie(ret, podHttpBody) = GetHttpBodyFromFlexVolume(url);
    if (ret != SUCCESS) {
        ERRLOG("Fail to call k8s API, uri: %s", url.c_str());
        return std::make_pair(ret, pvcVector);
    }
    YAML::Node nodeItems = YAML::Load(podHttpBody)["spec"]["volumes"];
    for (YAML::const_iterator nIt = nodeItems.begin(); nIt != nodeItems.end(); nIt++) {
        const YAML::Node &nodeItem = *nIt;
        std::string volumeName = nodeItem["name"].as<std::string>();
        if (std::find(volumeNames.begin(), volumeNames.end(), volumeName) != volumeNames.end()) {
            std::string pvcName = nodeItem["persistentVolumeClaim"]["claimName"].as<std::string>();
            Pv pv;
            pv.volumeName = volumeName;
            pv.pvcName = pvcName;
            pvcVector.push_back(pv);
        }
    }
    return std::make_pair(ret, pvcVector);
}

std::pair<int32_t, std::vector<Pv>> KubernetesPlugin::KubernetesApi::GetPvsByPvcNames(const std::vector<Pv> &pvcList)
{
    int ret = SUCCESS;
    std::vector<Pv> pvs = {};
    std::string podHttpBody;
    std::string url = config.GetClusterServer() + "/api/v1/persistentvolumes";
    std::tie(ret, podHttpBody) = GetHttpBodyFromFlexVolume(url);
    if (ret != SUCCESS) {
        ERRLOG("Fail to call k8s API, uri: %s", url.c_str());
        return std::make_pair(ret, pvs);
    }
    YAML::Node nodeItems = YAML::Load(podHttpBody)["items"];
    for (YAML::const_iterator nIt = nodeItems.begin(); nIt != nodeItems.end(); nIt++) {
        const YAML::Node &pvItem = *nIt;
        if (!pvItem["spec"]["claimRef"]) {
            continue;
        }
        std::string pvcName = pvItem["spec"]["claimRef"]["name"].as<std::string>();
        for (const auto &item: pvcList) {
            if (item.pvcName != pvcName) {
                continue;
            }
            Pv pv;
            pv.volumeName = item.volumeName;
            pv.pvcName = item.pvcName;
            pv.name = pvItem["metadata"]["name"].as<std::string>();
            pv.lunName = pvItem["spec"]["flexVolume"]["options"]["lunname"].as<std::string>();
            pv.size = pvItem["spec"]["capacity"]["storage"].as<std::string>();
            pv.storageUrl = pvItem["spec"]["flexVolume"]["options"]["url"].as<std::string>();
            pvs.push_back(pv);
        }
    }
    return std::make_pair(ret, pvs);
}

std::pair<int, std::vector<ApplicationResource>> KubernetesPlugin::KubernetesApi::ListStatefulSet(
    const Application &nameSpace)
{
    std::vector <ApplicationResource> stsResourceLists = {};
    int ret = 0;
    std::string stsName;
    bool isGetSingleSts = false;
    if (!nameSpace.extendInfo.empty()) {
        stsName = nameSpace.extendInfo;
        isGetSingleSts = true;
    }
    // 1. 预先将stateFulHttpBody、podsHttpBody、pvsHttpBody从FlexVolume中查出来。
    std::string stateFulHttpBody;
    std::string podsHttpBody;
    std::string pvsHttpBody;
    // 1.1 查statefulsets -> stateFulHttpBody
    std::string url = config.GetClusterServer() + "/apis/apps/v1/namespaces/" + nameSpace.name + "/statefulsets";
    if (isGetSingleSts) {
        url += "/" + stsName;
    }
    std::tie(ret, stateFulHttpBody) = GetHttpBodyFromFlexVolume(url);
    if (ret != SUCCESS) {
        ERRLOG("Get stateful set http body failed:%d", ret);
        return std::make_pair(ret, stsResourceLists);
    }
    // 1.2 查pods -> podsHttpBody
    url = config.GetClusterServer() + "/api/v1/namespaces/" + nameSpace.name + "/pods";
    std::tie(ret, podsHttpBody) = GetHttpBodyFromFlexVolume(url);
    if (ret != SUCCESS) {
        ERRLOG("Get pods http body failed:%d", ret);
        return std::make_pair(ret, stsResourceLists);
    }
    // 1.3 查pv -> pvsHttpBody
    url = config.GetClusterServer() + "/api/v1/persistentvolumes";
    std::tie(ret, pvsHttpBody) = GetHttpBodyFromFlexVolume(url);
    if (ret != SUCCESS) {
        ERRLOG("Get pvs http body failed:%d", ret);
        return std::make_pair(ret, stsResourceLists);
    }
    // 2. 基于stateFulHttpBody、podsHttpBody、pvsHttpBody填充相关StateFulSet结构体。
    return FulFillStsResourceList(stateFulHttpBody, podsHttpBody, pvsHttpBody, isGetSingleSts, nameSpace);
}

std::pair<int, std::vector<ApplicationResource>> KubernetesApi::FulFillStsResourceList(
    const std::string &stateFulHttpBody, const std::string &podsHttpBody, const std::string &pvsHttpBody,
    const bool &isGetSingleSts, const Application &nameSpace)
{
    std::vector <ApplicationResource> stsResourceLists;
    try {
        std::vector <StateFulSet> stateFulSetList{};
        if (isGetSingleSts) {
            stateFulSetList.push_back(GetSingleStatefulSetsFromHttpBody(stateFulHttpBody));
        } else {
            stateFulSetList = GetStatefulSetsFromHttpBody(stateFulHttpBody);
        }
        for (auto &stsItem: stateFulSetList) {
            stsItem.pods = GetPodsFromHttpBody(podsHttpBody, stsItem.name);
            for (auto &podItem: stsItem.pods) {
                FulfillPvsUnderPod(pvsHttpBody, podItem);
            }
            auto [ret, stsResource] = ConvertStateFulSetToAppResource(stsItem, nameSpace.id);
            if (ret != SUCCESS) {
                ERRLOG("Convert statefulSet to appResource failed");
                return std::make_pair(CONNECT_FAILED, stsResourceLists);
            }
            stsResourceLists.push_back(stsResource);
        }
        return std::make_pair(SUCCESS, stsResourceLists);
    } catch (std::exception &e) {
        ERRLOG("Illegal HttpBody Parse, message: %s", e.what());
        return std::make_pair(CONNECT_FAILED, stsResourceLists);
    }
}

std::string KubernetesApi::KubeExec(
    const std::string &nameSpace, const std::string &pod, const std::string &container, const std::string &command)
{
    // 构造WebSocket客户端双向认证需要的SSL Context
    Ssl::context ctx{Ssl::context::tlsv12_client};
    if (!config.SetSSLContext(ctx)) {
        return "";
    }

    // 获取K8S API Server的IP和端口
    std::string host;
    std::string port;
    if (!config.GetClusterServer(host, port)) {
        ERRLOG("Failed to get cluster server address");
        return "";
    }

    // 建立底层Socket连接
    Net::io_context ioc;
    Tcp::resolver resolver{ioc};
    Websocket::stream <Beast::ssl_stream<Tcp::socket>> ws{ioc, ctx};
    auto const results = resolver.resolve(host, port);
    auto ep = Net::connect(get_lowest_layer(ws), results);
    Defer _(nullptr, [&](...) {
        // 确保在后续分支退出后能关闭连接
        boost::system::error_code ec;
        ws.close(Websocket::close_code::normal, ec);
        INFOLOG("Close websocket finished, return code: %d", ec);
    });

    // 设置SNI主机名(很多主机需要这个信息才能握手成功)
    if (!SSL_set_tlsext_host_name(ws.next_layer().native_handle(), host.c_str())) {
        ERRLOG("Failed to set SNI hostname, error: %ld", ::ERR_get_error());
        return "";
    }
    host += ':' + std::to_string(ep.port());

    // 执行SSL的握手
    ws.next_layer().handshake(Ssl::stream_base::client);

    // 设置WebSoket握手消息中的User-Agent
    ws.set_option(Websocket::stream_base::decorator([](Websocket::request_type &req) {
        req.set(Http::field::user_agent, "backup-websocket-client");
        DBGLOG("Websocket request: %s", WIPE_SENSITIVE(req).c_str());
    }));

    // 执行WebSocket的握手（握手消息中包含kube exec API的调用）
    std::string req = CreateWebsocketRequest(nameSpace, pod, container, command);
    Websocket::response_type rsp;
    boost::system::error_code ec;
    ws.handshake(rsp, host, req, ec);
    if (ec) {
        ERRLOG("Websocket handshake failed, error code: %d", ec);
        return "OCI runtime exec failed: exec failed: " + ec.message();
    }
    DBGLOG("Websocket response: %s", WIPE_SENSITIVE(rsp).c_str());
    std::string res = GetWebsocketResponse(ws);
    DBGLOG("Kube exec result: %s ", WIPE_SENSITIVE(res).c_str());
    return res;
}

std::string KubernetesPlugin::KubernetesApi::GetWebsocketResponse(
        boost::beast::websocket::stream <boost::beast::ssl_stream<boost::asio::ip::tcp::socket>>
        &ws)
{
    // 读取命令执行结果
    Beast::flat_buffer buffer;
    boost::system::error_code ec;
    std::string cmdResult;
    std::size_t len = ws.read(buffer, ec);  // 第一次读取到的是K8s API Server返回的"STDOUT"，需忽略
    DBGLOG("First read, len: %d, body: %s", len, WIPE_SENSITIVE(Beast::buffers_to_string(buffer.data())).c_str());
    buffer.consume(len);
    while (1) {
        len = ws.read(buffer, ec);
        if (ec) {
            break;
        }
        cmdResult += Beast::buffers_to_string(buffer.data());
        buffer.consume(len);
        DBGLOG("While loop read, len: %d, body: %s,ec: %d", len,
               WIPE_SENSITIVE(Beast::buffers_to_string(buffer.data())).c_str(), ec);
    }
    return cmdResult;
}

HttpRequest KubernetesPlugin::KubernetesApi::CreateRequest(const std::string &method, const std::string &url)
{
    HttpRequest request;
    request.method = method;
    request.certCA = config.GetClusterCrt();
    request.clientCert = config.GetUserCrt();
    request.clientKey = config.GetUserKey();
    request.isVerify = Module::CACertVerification::AGENT_VERIFY;
    request.url = url;
    return request;
}

std::string KubernetesApi::CreateWebsocketRequest(
    const std::string &nameSpace, const std::string &pod, const std::string &container, const std::string &command)
{
    std::string assembleCmd = command;
    boost::trim(assembleCmd);                           // 去掉前后空格
    boost::replace_all(assembleCmd, " ", "&command=");  // 将命令中的空格替换成&command=以满足K8s API的要求
    std::map<std::string, std::string> pathParams = {
            {"namespace", nameSpace},
            {"pod",       pod},
            {"command",   assembleCmd},
            {"container", container}};

    // 检查参数是否合法
    bool f = std::any_of(pathParams.begin(), pathParams.end(), [](const auto &p) { return p.second.empty(); });
    if (f) {
        ERRLOG("Invalid input paramters");
        return "";
    }

    std::string res = "/api/v1/namespaces/{namespace}/pods/{pod}/"
                      "exec?command={command}&container={container}&stderr=true&stdout=true";
    for (const auto &path: pathParams) {
        boost::replace_all(res, "{" + path.first + "}", path.second);
    }

    DBGLOG("WebSocket request: %s", WIPE_SENSITIVE(res).c_str());
    return res;
}

std::vector<ApplicationResource> KubernetesPlugin::KubernetesApi::GetNameSpacesFromHttpBody(const std::string &stsBody,
    const std::string &namespaceBody)
{
    std::unordered_map<std::string, ApplicationResource> nameSpaceMap = {};
    std::vector<ApplicationResource> ret = {};

    YAML::Node items = YAML::Load(stsBody)["items"];
    for (YAML::const_iterator it = items.begin(); it != items.end(); it++) {
        const YAML::Node &statefulSet = *it;
        auto nameSpaceName = statefulSet["metadata"]["namespace"].as<std::string>();
        if (!statefulSet["spec"]["volumeClaimTemplates"]) {
            continue;
        }
        auto volumeNames = LoadVolumesNames(statefulSet["spec"]["volumeClaimTemplates"]);
        if (volumeNames.empty()) {
            continue;
        }
        ApplicationResource namespaceResource;
        namespaceResource.__set_id(config.GetClusterServer() + nameSpaceName);
        namespaceResource.__set_name(nameSpaceName);
        namespaceResource.__set_type("Namespace");
        namespaceResource.__set_subType("KubernetesNamespace");
        nameSpaceMap[nameSpaceName] = namespaceResource;
    }

    std::unordered_map<std::string, std::string> nsNameIdMap = {};
    YAML::Node nItems = YAML::Load(namespaceBody)["items"];
    for (YAML::const_iterator it = nItems.begin(); it != nItems.end(); it++) {
        const YAML::Node &namespaces = *it;
        auto name = namespaces["metadata"]["name"].as<std::string>();
        auto uid = namespaces["metadata"]["uid"].as<std::string>();
        nsNameIdMap[name] = uid;
    }
    ret.reserve(nameSpaceMap.size());
    for (const auto &x: nameSpaceMap) {
        std::string id = nsNameIdMap[x.first];
        ApplicationResource namespaceResource = x.second;
        namespaceResource.__set_id(id);
        ret.push_back(namespaceResource);
    }
    return ret;
}

std::vector<StateFulSet> KubernetesPlugin::KubernetesApi::GetStatefulSetsFromHttpBody(const std::string &stsBody)
{
    std::vector<StateFulSet> statefulSetVector = {};
    YAML::Node items = YAML::Load(stsBody)["items"];
    for (YAML::const_iterator it = items.begin(); it != items.end(); it++) {
        const YAML::Node &statefulSet = *it;
        if (!statefulSet["spec"]["volumeClaimTemplates"]) {
            continue;
        }
        StateFulSet statefulSetResource;
        statefulSetResource.id = statefulSet["metadata"]["uid"].as<std::string>();
        statefulSetResource.name = statefulSet["metadata"]["name"].as<std::string>();
        statefulSetResource.nameSpace = statefulSet["metadata"]["namespace"].as<std::string>();
        statefulSetResource.replicasNum = statefulSet["spec"]["replicas"].as<int>();
        statefulSetResource.volumeNames = LoadVolumesNames(statefulSet["spec"]["volumeClaimTemplates"]);
        if (!statefulSetResource.volumeNames.empty()) {
            statefulSetVector.push_back(statefulSetResource);
        }
    }
    return statefulSetVector;
}

StateFulSet KubernetesPlugin::KubernetesApi::GetSingleStatefulSetsFromHttpBody(const std::string &stsBody)
{
    YAML::Node statefulSet = YAML::Load(stsBody);
    StateFulSet statefulSetResource;
    statefulSetResource.id = statefulSet["metadata"]["uid"].as<std::string>();
    statefulSetResource.name = statefulSet["metadata"]["name"].as<std::string>();
    statefulSetResource.nameSpace = statefulSet["metadata"]["namespace"].as<std::string>();
    statefulSetResource.replicasNum = statefulSet["spec"]["replicas"].as<int>();
    statefulSetResource.volumeNames = LoadVolumesNames(statefulSet["spec"]["volumeClaimTemplates"]);
    return statefulSetResource;
}

std::vector<Pod> KubernetesApi::GetPodsFromHttpBody(const std::string &podBody, const std::string &stsName)
{
    std::vector<Pod> podVector = {};
    YAML::Node nodeItems = YAML::Load(podBody)["items"];
    for (YAML::const_iterator nIt = nodeItems.begin(); nIt != nodeItems.end(); nIt++) {
        const YAML::Node &nodeItem = *nIt;
        auto podName = nodeItem["metadata"]["name"].as<std::string>();
        if (std::string::npos == podName.rfind(stsName, 0)) {
            continue;
        }
        Pod pod;
        pod.name = podName;
        YAML::Node containers = nodeItem["spec"]["containers"];
        if (!containers || containers.size() == 0) {
            ERRLOG("There is no any container in the POD, name: %s", podName.c_str());
            continue;
        }
        for (const auto &container: containers) {
            auto containerName = container["name"].as<std::string>();
            pod.containerNames.push_back(containerName);
        }
        YAML::Node volumeItems = nodeItem["spec"]["volumes"];
        for (YAML::const_iterator vIt = volumeItems.begin(); vIt != volumeItems.end(); vIt++) {
            const YAML::Node &volumeItem = *vIt;
            if (volumeItem["persistentVolumeClaim"]) {
                Pv pv;
                pv.volumeName = volumeItem["name"].as<std::string>();
                pv.pvcName = volumeItem["persistentVolumeClaim"]["claimName"].as<std::string>();
                pod.pvs.push_back(pv);
            }
        }
        podVector.push_back(pod);
    }

    return podVector;
}

std::pair<int, std::string> KubernetesPlugin::KubernetesApi::GetHttpBodyFromFlexVolume(const std::string &url)
{
    HttpRequest request = CreateRequest("Get", url);

    HttpResponseInfo response;
    int32_t ret = client.SendRequest(request, response);
    if (ret == SUCCESS && response.m_statusCode == Module::SC_OK) {
        return std::make_pair(SUCCESS, response.m_body);
    } else {
        return std::make_pair(FAILED, response.m_body);
    }
}

std::pair<int, ApplicationResource> KubernetesApi::ConvertStateFulSetToAppResource(StateFulSet &stateFulSet,
                                                                                   const std::string &nameSpaceId)
{
    ApplicationResource stsResource;
    stsResource.__set_id(stateFulSet.id);
    stsResource.__set_name(stateFulSet.name);
    stsResource.__set_parentName(stateFulSet.nameSpace);
    stsResource.__set_parentId(nameSpaceId);
    stsResource.__set_type("StatefulSet");
    stsResource.__set_subType("KubernetesStatefulSet");
    // 序列化ExtendInfo
    StateFulSetExtend stsExtendInfo;
    Module::JsonHelper::StructToJsonString(stateFulSet, stsExtendInfo.sts);
    Module::JsonHelper::StructToJsonString(stsExtendInfo, stsResource.extendInfo);
    stsResource.__set_extendInfo(stsResource.extendInfo);
    return std::make_pair(SUCCESS, stsResource);
}

void KubernetesPlugin::KubernetesApi::FulfillPvsUnderPod(const std::string &pvBody, KubernetesPlugin::Pod &pod)
{
    YAML::Node pvItems = YAML::Load(pvBody)["items"];
    for (YAML::const_iterator nIt = pvItems.begin(); nIt != pvItems.end(); nIt++) {
        const YAML::Node &pvItem = *nIt;
        auto pvName = pvItem["metadata"]["name"].as<std::string>();
        if (!pvItem["spec"]["claimRef"]) {
            continue;
        }
        for (auto &pv: pod.pvs) {
            if (pv.pvcName == pvItem["spec"]["claimRef"]["name"].as<std::string>()) {
                pv.name = pvItem["metadata"]["name"].as<std::string>();
                pv.lunName = pvItem["spec"]["flexVolume"]["options"]["lunname"].as<std::string>();
                pv.size = pvItem["spec"]["capacity"]["storage"].as<std::string>();
                pv.storageUrl = pvItem["spec"]["flexVolume"]["options"]["url"].as<std::string>();
                break;
            }
        }
    }
}


int KubernetesApi::StopStateFulSet(const StateFulSet &sts)
{
    return SetReplicas(sts, 0);
}

int KubernetesApi::RestoreStateFulSet(const StateFulSet &sts)
{
    return SetReplicas(sts, sts.replicasNum);
}

int KubernetesPlugin::KubernetesApi::SetReplicas(const KubernetesPlugin::StateFulSet &sts, int val)
{
    // PATCH /apis/apps/v1/namespaces/{namespace}/statefulsets/{name}
    std::string url = config.GetClusterServer() + "/apis/apps/v1/namespaces/" +
                      sts.nameSpace + "/statefulsets/" + sts.name;
    HttpRequest request = CreateRequest("PATCH", url);

    request.heads.insert({"Content-Type", "application/json-patch+json"});
    std::string patchBody = R"([{"op": "replace", "path": "/spec/replicas", "value": )" + std::to_string(val) + "}]";
    request.body = patchBody;

    HttpResponseInfo response;
    int32_t ret = client.SendRequest(request, response);
    if (ret == SUCCESS && response.m_statusCode == Module::SC_OK) {
        return SUCCESS;
    } else {
        ERRLOG("SetReplicas failed to FlexVolume");
        return FAILED;
    }
}

std::pair<int, std::set<std::string>> KubernetesApi::ListStorages()
{
    std::set<std::string> storageUrlSet = {};
    std::string url = config.GetClusterServer() + "/api/v1/persistentvolumes";
    auto [ret, pvBody] = GetHttpBodyFromFlexVolume(url);
    if (ret != SUCCESS) {
        return std::make_pair(ret, storageUrlSet);
    }

    YAML::Node pvItems = YAML::Load(pvBody)["items"];
    for (YAML::const_iterator nIt = pvItems.begin(); nIt != pvItems.end(); nIt++) {
        const YAML::Node &pvItem = *nIt;
        if (pvItem["spec"]["flexVolume"]) {
            auto storageUrl = pvItem["spec"]["flexVolume"]["options"]["url"].as<std::string>();
            storageUrlSet.insert(storageUrl);
        }
    }
    return std::make_pair(ret, storageUrlSet);
}
}